from pyspark import SparkContext, SparkConf
from datetime import datetime
from pyecharts import options as opts
from pyecharts.charts import Bar, Line

def parse_line(line):
    try:
        fields = line.split(',')
        return (
            fields[0],  # 用户ID
            fields[1],  # 商品ID
            fields[2],  # 商品类别
            fields[3],  # 行为类型
            datetime.strptime(fields[4], "%Y-%m-%d %H:%M:%S"),  # 时间戳
            fields[5],  # 地域
            float(fields[6])  # 商品价格
        )
    except Exception as e:
        print(f"解析错误: {e}")
        return None

if __name__ == "__main__":
    conf = SparkConf().setAppName("EcommerceAnalysis")
    sc = SparkContext(conf=conf)
    sc.setLogLevel("WARN")  # 减少INFO日志

    # 读取数据（使用绝对路径）
    raw_data = sc.textFile("D://ecommerce_env/user_behavior.csv")
    print("数据总行数:", raw_data.count())  # 关键调试点

    # 跳过表头
    header = raw_data.first()
    parsed_data = raw_data.filter(lambda line: line != header).map(parse_line).filter(lambda x: x is not None).cache()
    print("有效数据行数:", parsed_data.count())  # 关键调试点

    # 指标1：行为类型分布
    behavior_counts = parsed_data.map(lambda x: (x[3], 1)).reduceByKey(lambda a, b: a + b).collect()
    print("行为类型分布:", behavior_counts)

    # 指标2：商品类别点击量Top10
    click_data = parsed_data.filter(lambda x: x[3] == "点击")
    category_clicks = click_data.map(lambda x: (x[2], 1)).reduceByKey(lambda a, b: a + b).sortBy(lambda x: -x[1]).take(10)
    print("商品类别点击量Top10:", category_clicks)

    # 指标3：各省份购买用户数
    purchase_users = parsed_data.filter(lambda x: x[3] == "购买").map(lambda x: (x[5], x[0])).distinct()
    province_purchase_count = purchase_users.map(lambda x: (x[0], 1)).reduceByKey(lambda a, b: a + b).collect()
    print("各省份购买用户数:", province_purchase_count)

    # 指标4：小时级活跃用户数
    hour_active = parsed_data.map(lambda x: (x[4].hour, x[0]))
    hour_unique_users = hour_active.groupByKey().mapValues(lambda ids: len(set(ids))).collect()
    print("小时级活跃用户数:", hour_unique_users)

    # 使用 pyecharts 可视化行为类型分布
    behavior_types = [item[0] for item in behavior_counts]
    behavior_counts_list = [item[1] for item in behavior_counts]
    bar_behavior = (
        Bar()
        .add_xaxis(behavior_types)
        .add_yaxis("行为数量", behavior_counts_list)
        .set_global_opts(title_opts=opts.TitleOpts(title="行为类型分布"))
    )

    # 可视化商品类别点击量Top10
    category_names = [item[0] for item in category_clicks]
    category_click_counts = [item[1] for item in category_clicks]
    bar_category = (
        Bar()
        .add_xaxis(category_names)
        .add_yaxis("点击量", category_click_counts)
        .set_global_opts(title_opts=opts.TitleOpts(title="商品类别点击量Top10"))
    )

    # 可视化各省份购买用户数
    provinces = [item[0] for item in province_purchase_count]
    purchase_counts = [item[1] for item in province_purchase_count]
    bar_province = (
        Bar()
        .add_xaxis(provinces)
        .add_yaxis("购买用户数", purchase_counts)
        .set_global_opts(title_opts=opts.TitleOpts(title="各省份购买用户数"))
    )

    # 可视化小时级活跃用户数
    hours = [item[0] for item in hour_unique_users]
    user_counts = [item[1] for item in hour_unique_users]
    line_hour = (
        Line()
        .add_xaxis(hours)
        .add_yaxis("活跃用户数", user_counts)
        .set_global_opts(title_opts=opts.TitleOpts(title="小时级活跃用户数"))
    )

    # 将多个图表组合到一个 HTML 文件中
    page = bar_behavior.overlap(bar_category).overlap(bar_province).overlap(line_hour)
    page.render("ecommerce_analysis.html")

    sc.stop()

